#include "buffer/buffer_pool_manager.h"

#include <memory>

#include "buffer/page_allocator.h"

namespace boltDB {

BufferPoolManager::BufferPoolManager(size_t pool_size,
                                     BasicPageAllocator *allocator,
                                     DiskManager *disk_manager,
                                     size_t replacer_k)
    : pool_size_(pool_size),
      page_allocator_(allocator),
      disk_scheduler_(std::make_unique<DiskScheduler>(disk_manager)) {
  // 分配缓冲池
  pages_ = new Page[pool_size_];
  replacer_ = std::make_unique<LRUKReplacer>(pool_size, replacer_k);
  // 初始化空闲链表.
  for (size_t i = 0; i < pool_size_; ++i) {
    free_list_.emplace_back(static_cast<int>(i));
  }
}
BufferPoolManager::~BufferPoolManager() { delete[] pages_; }

auto BufferPoolManager::NewPage(page_id_t *page_id) -> Page * {
  std::scoped_lock<std::mutex> locker(latch_);
  // 首先看空闲链表，如果有空的位置可以缓存页面，就直接缓存就行，
  // 否则缓冲池已满，就要驱逐出一个页面，给新页面腾地方,如果缓冲区
  // 所有的页面都不可驱逐，那么就要返回空指针
  int frame_id{-1};
  if (!free_list_.empty()) {
    frame_id = free_list_.front();
    free_list_.pop_front();
  } else if (!replacer_->Evict(&frame_id)) {
    return nullptr;
  }

  Page *page = pages_ + frame_id;
  // 如果被换出内存的页是个脏页，就要先写回磁盘同步数据后再清空
  if (page->GetPageId() != INVALID_PAGE_ID && page->IsDirty()) {
    auto promise = disk_scheduler_->CreatePromise();
    auto future = promise.get_future();
    disk_scheduler_->Schedule({true, page->data_, page->page_id_, std::move(promise)});
    future.get();
    page->is_dirty_ = false;
  }
  // 分配新的page_id,删除旧的对应关系
  *page_id = page_allocator_->AllocatePage();
  page_table_.erase(page->page_id_);
  page_table_.emplace(*page_id, frame_id);
  // 初始化页的相关信息
  page->page_id_ = *page_id;
  page->ref_count_ = 1;
  page->ResetMemory();
  // 由于页被引用了，所以不可被驱逐出去
  replacer_->RecordAccess(frame_id);
  replacer_->SetEvictable(frame_id, false);
  return page;
}

auto BufferPoolManager::FetchPage(page_id_t page_id) -> Page * {
  if (page_id == INVALID_PAGE_ID) {
    return nullptr;
  }
  std::scoped_lock<std::mutex> locker(latch_);
  // 首先看缓冲池里面有没有，有的话就增加引用计数，返回就行
  // 没有的话就从磁盘读一个页出来
  auto it = page_table_.find(page_id);
  if (it != page_table_.end()) {
    auto page = pages_ + it->second;
    page->ref_count_++;
    replacer_->RecordAccess(it->second);
    replacer_->SetEvictable(it->second, false);
    return page;
  }
  frame_id_t frame_id{-1};
  if (!free_list_.empty()) {
    frame_id = free_list_.front();
    free_list_.pop_front();
  } else if (!replacer_->Evict(&frame_id)) {
    return nullptr;
  }
  Page *page = pages_ + frame_id;
  // 如果被换出内存的页是个脏页，就要先写回磁盘同步数据后再清空
  if (page->page_id_ != INVALID_PAGE_ID && page->IsDirty()) {
    auto result = disk_scheduler_->CreatePromise();
    auto future = result.get_future();
    disk_scheduler_->Schedule({true, page->data_, page->page_id_, std::move(result)});
    future.get();
    page->is_dirty_ = false;
    page->ref_count_ = 0;
  }
  // 从磁盘中读页
  page_table_.erase(page->page_id_);
  page_table_.emplace(page_id, frame_id);
  page->page_id_ = page_id;
  page->ref_count_ = 1;
  page->ResetMemory();
  auto result = disk_scheduler_->CreatePromise();
  auto future = result.get_future();
  disk_scheduler_->Schedule({false, page->data_, page->page_id_, std::move(result)});
  future.get();
  replacer_->RecordAccess(frame_id);
  replacer_->SetEvictable(frame_id, false);
  return page;
}

auto BufferPoolManager::UnRefPage(page_id_t page_id, bool is_dirty) -> bool {
  std::scoped_lock<std::mutex> locker(latch_);
  // 先找缓冲池中有没有要找的页，没有的话就返回
  // 有的话就更新一下引用计数
  auto it = page_table_.find(page_id);
  if (it == page_table_.end()) {
    return false;
  }
  auto frame_id = it->second;
  auto page = pages_ + frame_id;
  page->is_dirty_ |= is_dirty;
  if (page->ref_count_ == 0) {
    return false;
  }
  page->ref_count_--;
  if (page->ref_count_ == 0) {
    replacer_->SetEvictable(frame_id, true);
  }
  return true;
}

auto BufferPoolManager::FlushPage(page_id_t page_id) -> bool {
  std::scoped_lock<std::mutex> locker(latch_);
  auto it = page_table_.find(page_id);
  if (it == page_table_.end()) {
    return false;
  }
  auto frame_id = it->second;
  auto page = pages_ + frame_id;
  if (page->page_id_ == INVALID_PAGE_ID) {
    return false;
  }
  auto promise = disk_scheduler_->CreatePromise();
  auto future = promise.get_future();
  disk_scheduler_->Schedule({true, page->data_, page->page_id_, std::move(promise)});
  future.get();
  page->is_dirty_ = false;
  return true;
}

void BufferPoolManager::FlushAllPages() {
  std::scoped_lock<std::mutex> locker(latch_);
  for (auto &[page_id, frame_id] : page_table_) {
    if (pages_[frame_id].page_id_ == INVALID_PAGE_ID) {
      continue;
    }
    auto promise = disk_scheduler_->CreatePromise();
    auto future = promise.get_future();
    disk_scheduler_->Schedule({true, pages_[frame_id].data_, pages_[frame_id].page_id_, std::move(promise)});
    future.get();
    pages_[frame_id].is_dirty_ = false;
  }
}

auto BufferPoolManager::DeletePage(page_id_t page_id) -> bool {
  // 先找缓冲池中有没有要找的页，没有的话就返回 true
  std::scoped_lock<std::mutex> locker(latch_);
  auto it = page_table_.find(page_id);
  if (it == page_table_.end()) {
    return true;
  }
  auto frame_id = it->second;
   // 如果页面还在被使用，就不要删除
  if (pages_[frame_id].ref_count_ > 0) {
    return false;
  }
  page_table_.erase(page_id);
  replacer_->Remove(frame_id);
  free_list_.push_back(frame_id);
  page_allocator_->DeallocatePage(page_id);
  // 清空相关属性及元数据
  pages_[frame_id].ResetMemory();
  pages_[frame_id].is_dirty_ = false;
  pages_[frame_id].ref_count_ = 0;
  pages_[frame_id].page_id_ = INVALID_PAGE_ID;
  return true;
}

auto BufferPoolManager::FetchPageBasic(page_id_t page_id) -> BasicPageGuard {
  auto page = FetchPage(page_id);
  return {this, page};
}

auto BufferPoolManager::FetchPageRead(page_id_t page_id) -> ReadPageGuard {
  auto basic_guard = FetchPageBasic(page_id);
  return basic_guard.UpgradeRead();
}

auto BufferPoolManager::FetchPageWrite(page_id_t page_id) -> WritePageGuard {
  auto basic_guard = FetchPageBasic(page_id);
  return basic_guard.UpgradeWrite();
}

auto BufferPoolManager::NewPageGuarded(page_id_t *page_id) -> BasicPageGuard {
  auto page = NewPage(page_id);
  return {this, page};
}

}  // namespace boltDB